package realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import realtime.util.EnvUtil;
import realtime.util.KafkaUtil;

public class DwdInteractionCourseReview {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = EnvUtil.getSEE(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.executeSql(KafkaUtil.getTopicDbDDL("dwd_interaction_course_review_group1"));
        Table reviewInfo = tableEnv.sqlQuery("select\n" +
                "    `data`['id'] id,\n" +
                "    `data`['user_id'] user_id,\n" +
                "    `data`['course_id'] course_id,\n" +
                "    `data`['review_stars'] review_stars,\n" +
                "    `data`['create_time'] create_time,\n" +
                "    ts\n" +
                "from topic_db where `table`='review_info' and `type`='insert'");
        tableEnv.createTemporaryView("review_info",reviewInfo);

        //tableEnv.executeSql(" select * from review_info").print();

        tableEnv.executeSql("CREATE TABLE dwd_interaction_course_review (\n" +
                "  id string,\n" +
                "  user_id  string,\n" +
                "  course_id string,\n" +
                "  review_stars string,\n" +
                "  create_time string,\n" +
                "  ts  string,\n" +
                "  PRIMARY KEY (id) NOT ENFORCED\n" +
                ") " + KafkaUtil.getUpsertKafkaDDL("dwd_interaction_course_review"));
        tableEnv.executeSql("insert into dwd_interaction_course_review select * from review_info");
    }
}
